﻿/**
  ******************************************************************************
  * @file           : WakeupQueue.h
  * @author         : livio[yuanfeng1897@outlook.com]
  * @brief          : None
  * @attention      : None
  * @date           : 2023/3/6
  ******************************************************************************
  */



#ifndef CPPDEMO_WAKEUPQUEUE_H
#define CPPDEMO_WAKEUPQUEUE_H

#include "IIOObject.h"
#include "IIOManager.h"

#include <mutex>
#include <queue>
#include <vector>
#include <stdexcept>
#include <functional>


/**
 * 定义一个唤醒队列
 * 利用socketpair创建一对socket 句柄。
 * 通过这一对sockfd模拟管道通信，达到唤醒效果
 * Linux下可以直接创建pipe或者使用socketpair创建一对相互通信的socket来实现唤醒效果
 * Windows下需要实现Linux下函数socketpair。利用socket实现唤醒效果
 * 因为windows中pipe一类的句柄是不被io多路复用（select/poll）支持的
 * 所以使用socket模拟pipe行为。
 */
template<class T>
class WakeupQueue : IIOObject {
    enum class IOTYPE : uint8_t {
        WRITE = 0,
        READ = 1,
    };
public:
    /**
     * constructor
     * @param _iomanager  io事件循环对象。管理IO的注册注销修改事件
     * @param onWakeup    回调函数。有事件时通过该回调函数通知
     */
    WakeupQueue(IIOManager *_iomanager, std::function<void(T)> onWakeup, uint32_t max = UINT16_MAX)
            : maxSize{max}, m_iomanager{_iomanager}, m_onWakeup{onWakeup} {
        auto err = socketpair(AF_UNIX, SOCK_STREAM, 0, sockPair);
        if (err < 0) throw std::runtime_error{"call func[socketpair] failed."};

        if (m_iomanager)
            m_iomanager->registerEvent(*this, IOEvent{IOEvent::Event::RDWD});
    }

    /**
     * destructor
     */
    ~WakeupQueue() {
        if (m_iomanager)
            m_iomanager->unregisterEvent(*this);
    }

    [[nodiscard]] const SOCKET getFileDescriptor() const override {
        return sockPair[IOTYPE::READ];
    }

    bool push(const T &t) {
        {
            std::lock_guard<std::mutex> locker{m_mtxQueue};
            m_queue.push(t);
        }
        // 发送多少字节，表示有多少个事件
        char buf[1] = {1};
        return ::send(sockPair[IOTYPE::WRITE], buf, 1) > 0;
    }

    bool push(T &&t) {
        {
            std::lock_guard<std::mutex> locker{m_mtxQueue};
            m_queue.push(std::forward<T>(t));
        }
        // 发送多少字节，表示有多少个事件
        char buf[1] = {1};
        return ::send(sockPair[IOTYPE::WRITE], buf, 1, 0) > 0;
    }

protected:
    void onReadEvent() override {
        char buf[1] = {1};
        auto recvBytes = ::recv(sockPair[IOTYPE::READ], buf, 1, 0);
        // 发送多少字节，表示有多少个事件；接收的字节数就是queue里面的事件个数
        if (recvBytes <= 0) return;
        {
            std::lock_guard<std::mutex> locker{m_mtxQueue};
            for (int i = 0; i < recvBytes; ++i) {
                if (m_queue.empty()) {
                    // 不在这里调用回调。因为不知道回调耗时。
                    m_popStore.emplace_back(m_queue.front());
                    m_queue.pop();
                }
            }
        }
        // 调用回调函数也不需要加锁
        // 统计完有多少事件。依次调用回调
        for (auto &&val: m_popStore) {
            m_onWakeup(std::forward<T>(val));
        }
        m_popStore.clear();
    }

    void onWriteEvent() override {}

private:
    SOCKET sockPair[2];

    uint32_t maxSize{INT16_MAX};
    IIOManager *m_iomanager{nullptr};
    std::function<void(T)> m_onWakeup{nullptr};

    std::mutex m_mtxQueue;
    std::queue<T> m_queue;
    std::vector<T> m_popStore;
};

#endif //!CPPDEMO_WAKEUPQUEUE_H
